package com.ht.api.call.func.impl.cdc;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.ht.api.bean.AjaxResult;
import com.ht.api.bean.LoginBean;
import com.ht.api.call.func.IFunctionCall;
import com.ht.api.db.Db;
import com.ht.api.db.Entity;
import com.ht.api.db.exception.HtException;
import com.ht.api.db.hlink.JsonTreeProcessor;
import com.ht.api.db.hlink.JsonTreeProcessor.JsonTreeNode;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;

/**
 * 刷新CDC数据源
 * @author asq
 * @createTime 2025年3月10日 10:23:09
 */
@Slf4j
public class RefreshCDCFunctionCall implements IFunctionCall{

	@Override
	public String getName() {
		return "刷新CDC数据源";
	}

	@Override
	public JSONObject execute(LoginBean me, JSONObject param, AjaxResult result) {
		// 查询open_app表数据
		String mainId = param.getStr("main_id");
		if (StrUtil.isBlank(mainId)) {
			throw new HtException("main_id不能为空");
		}
		Entity openApp = Db.createSqlParser("open_app").findById(mainId);
		if (openApp == null) {
			throw new HtException("open_app不存在，main_id：" + mainId);
		}
		
		// 刷新cdc对应的hlink数据源
		String configStr = openApp.getStr("config");
		long tenantId = openApp.getLong("tenant_id");
		if (StrUtil.isNotBlank(configStr)) {
			JSONObject appConfig = JSONUtil.parseObj(configStr);
			JSONObject config = appConfig.getJSONObject("cdc_config");
			String hostname = config.getStr("hostname");
			int port = config.getInt("port");
			String username = config.getStr("username");
			String password = config.getStr("password");
			String database = config.getStr("database");
			List<String> tableArr = config.getBeanList("table", String.class);
			
			// 获取目标数据源的元数据
			Map<String, Entity> mapCloumn = getMetaList(mainId, hostname, port, username, password, database, tableArr);
			Map<String, String> mapTable = getTableNameList(hostname, port, username, password, database, tableArr);
			
			// 根据cdc配置，生成HlinkSource配置数据
			List<Entity> hlinkSourceList = new ArrayList<>();
			tableArr.stream().forEach(table -> {
				Entity hlinkSource = autoConfigHlinkSource(tenantId, mapTable, mapCloumn, database, table);
				hlinkSourceList.add(hlinkSource);
			});
			
			// 批量保存
			Db.insertBatchWithStream(hlinkSourceList);
			log.info("变更cdc配置，自动配置HlinkSource，共{}条", hlinkSourceList.size());
		} else {
			throw new HtException("cdc未正确配置，刷新数据源失败");
		}
		
		return JSONUtil.createObj();
	}

	/**
	 * 创建Hlink数据源
	 * @param tenantId 租户ID
	 * @param mapTable 数据库下面所有需要映射的表的结构信息
	 * @param mapCloumn 数据库下面所有需要映射的表的结构信息
	 * @param database 数据库名称
	 * @param table 表名称
	 * @return
	 */
	private Entity autoConfigHlinkSource(long tenantId, Map<String, String> mapTable, 
			Map<String, Entity> mapCloumn, String database, String table) {
		Entity entity = mapCloumn.get(database + ":" + table);
		String name = mapTable.get(database + ":" + table);
		String sourceCode = String.format("cdc_%s_%s_%s", tenantId, database, table);
		JsonTreeNode tree = JsonTreeProcessor.parseJsonToTree(entity.toString());
		
		// 根据sourceCode查询是否已存在
		Entity hlinkSource = Db.createSqlParser("hlink_source").where("code", sourceCode).findOne();
		if (hlinkSource == null) {
			hlinkSource = Entity.create("hlink_source");
			hlinkSource.set("code", sourceCode);
			hlinkSource.set("group_id", 1898999096174317568L);
		}
		hlinkSource.set("name", String.format("%s_%s", name, table));
		hlinkSource.set("source_json", entity.toString());
		hlinkSource.set("tree_json", JSONUtil.parseObj(tree).toString());
		
		return hlinkSource;
	}
	
	/**
     * 根据数据库连接信息，读取指定表的中文注释名称
     * @param hostname 数据库 IP
     * @param port     数据库端口
     * @param username 数据库用户名
     * @param password 数据库密码
     * @param database 数据库名称
     * @param tables   需要读取的表名列表
     */
    private static Map<String, String> getTableNameList(String hostname, int port, String username, String password,
                            String database, List<String> tables) {
        // 数据库连接 URL
        String url = "jdbc:mysql://" + hostname + ":" + port + "/" + database + "?useSSL=false&serverTimezone=UTC";

        // 存储所有表的元数据
        try (Connection connection = DriverManager.getConnection(url, username, password);
            Statement statement = connection.createStatement()) {
        	Map<String, String> map = new HashMap<>();

            // 遍历每个表
            for (String table : tables) {
                // 查询 information_schema.TABLES 表获取表注释
                String sql = "SELECT table_comment FROM information_schema.TABLES WHERE TABLE_SCHEMA=? AND TABLE_NAME=?";
                
                String tableName = "";
                try (PreparedStatement pstmt = connection.prepareStatement(sql)) {
                    pstmt.setString(1, database);
                    pstmt.setString(2, table);

                    try (ResultSet rs = pstmt.executeQuery()) {
                        while (rs.next()) {
                            tableName = rs.getString("table_comment");
                        }
                    }
                }
                
                map.put(database + ":" + table, tableName);
            }

            return map;
        } catch (Exception e) {
        	log.error("变更cdc配置，自动配置HlinkSource，获取表结构失败:{}", e.getMessage(), e);
        	throw new HtException("变更cdc配置，自动配置HlinkSource，获取表结构失败");
        }
    }
	
	/**
     * 根据数据库连接信息，读取指定表的结构并生成 JSON 集合
	 * @param appId 应用ID
	 * @param hostname 数据库 IP
	 * @param port     数据库端口
	 * @param username 数据库用户名
	 * @param password 数据库密码
	 * @param database 数据库名称
	 * @param tables   需要读取的表名列表
     */
    private static Map<String, Entity> getMetaList(String appId, String hostname, int port, String username,
                            String password, String database, List<String> tables) {
        // 数据库连接 URL
        String url = "jdbc:mysql://" + hostname + ":" + port + "/" + database + "?useSSL=false&serverTimezone=UTC";

        // 存储所有表的元数据
        try (Connection connection = DriverManager.getConnection(url, username, password);
            Statement statement = connection.createStatement()) {
        	Map<String, Entity> map = new HashMap<>();

            // 遍历每个表
            for (String table : tables) {
                // 查询表结构
            	Entity entity = Entity.create();
                // 查询 information_schema.COLUMNS 表获取字段注释
                String sql = "SELECT COLUMN_NAME, COLUMN_COMMENT " +
                             "FROM information_schema.COLUMNS " +
                             "WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ?";
                
                try (PreparedStatement pstmt = connection.prepareStatement(sql)) {
                    pstmt.setString(1, database);
                    pstmt.setString(2, table);

                    try (ResultSet rs = pstmt.executeQuery()) {
                        while (rs.next()) {
                            String columnName = rs.getString("COLUMN_NAME");
                            String columnComment = rs.getString("COLUMN_COMMENT");
                            entity.set(columnName, columnComment);
                        }
                    }
                }
                
                // 组装一个大的对象
            	Entity json = Entity.create();
            	json.set("appid", "应用ID");
                json.set("type", "操作类型");
                json.set("database", "数据库名");
                json.set("table", "表名");
                json.set("time", "时间戳");
            	//json.set("before", entity);
            	json.set("after", entity);
                
                map.put(database + ":" + table, json);
            }

            return map;
        } catch (Exception e) {
        	log.error("变更cdc配置，自动配置HlinkSource，获取表结构失败:{}", e.getMessage(), e);
        	throw new HtException("变更cdc配置，自动配置HlinkSource，获取表结构失败");
        }
    }
}
